/*
 * queues.h
 *
 *  (c) Normen Seemann
 *  Created on: May 2, 2009
 *
 */

#ifndef QUEUES_H_
#define QUEUES_H_

#include <stdlib.h>
#include <vector>

#include <boost/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/locks.hpp>
#include <boost/thread/condition_variable.hpp>

#include <core/data.h>
#include <core/atomic.h>

namespace crossEyed
{
namespace core
{

template<class REP>
class TDataQueue
{
protected:
   size_t N;

   //
   // Mutex and condition variable to get notified on new frame arrival
   //
   boost::mutex              newFrameMutex;
   boost::condition_variable newFrameCond;

   //
   // Mutex and condition variable to get notified when subscriptions get unsubscribed
   //
   boost::mutex              unsubscribeMutex;
   boost::condition_variable unsubscribeCond;

public:

   const static int InvalidSlot = -1 ;

   class TSubscription
   {
   public:
      REP * pRep ;
      int   slot ;

      TSubscription() : pRep( NULL ), slot ( InvalidSlot ) { } ;
      TSubscription(REP * pRep, int slot) : pRep( pRep ), slot( slot ) { } ;
   };

protected:

   //
   // Nested classes
   //
   class TData
   {
   public:
      REP *      pRep ;
      AtomicUInt numReaders ;

      TData() : pRep( NULL ), numReaders( 0 ) { } ;

   };

   //
   // Typedefs
   //
   typedef std::vector<TData> TDataVector ;

   TDataVector  dataVector ;

   //
   // This is the slot number a subscriber should get at any given time.
   //
   AtomicUInt   writerDataSlot ;
   int          readerDataSlot ;

public:

   TDataQueue( unsigned int N )
      : N(N),
        dataVector(N),
        writerDataSlot( 0 ),
        readerDataSlot( InvalidSlot )
        { } ;

   ~TDataQueue() { } ;

   inline size_t getN()
   {
      return N ;
   }

   inline unsigned int getNextDataSlot( unsigned int dataSlot )
   {
      return ( dataSlot + 1 ) % N  ;
   }

   inline void drainNextWriterDataSlot()
   {
      //
      // Set new readerDataSlot - note, it may wrap
      //
      unsigned int nextWriterDataSlot = getNextDataSlot( writerDataSlot.peek() ) ;

      TData & data = dataVector[ nextWriterDataSlot ] ;

      //
      // Wait for the next frame to drain.
      //
      boost::unique_lock<boost::mutex> lock( unsubscribeMutex );
      while( data.numReaders > 0 )
      {
         try
         {
            boost::this_thread::interruption_point() ;
         }
         catch( ... )
         {
            lock.unlock();
            throw;
         }

         unsubscribeCond.wait(lock);
      }
   }

   inline void advanceWriterDataSlot()
   {
      unsigned int oldWriterDataSlot = writerDataSlot.peek();

      //
      // We assume there is only one writer.
      //
      {
         boost::lock_guard<boost::mutex> lock( newFrameMutex );

         writerDataSlot.poke( getNextDataSlot( oldWriterDataSlot ) ) ;

         readerDataSlot = oldWriterDataSlot ;
      }

      newFrameCond.notify_all();
   }

   inline REP * getWriterData()
   {
      //
      // This routine is only safe if called from the writer thread.
      //
      return dataVector[ writerDataSlot ].pRep ;
   }

   inline REP * swizzleWriterData( REP * pRep )
   {
      //
      // This routine is only safe if called from the writer thread.
      //
      TData & data = dataVector[ writerDataSlot ] ;
      REP * pTempRepresentation = data.pRep ;
      data.pRep = pRep ;

      return pTempRepresentation ;
   }

   void writerDataFinishAndAdvance()
   {
      //
      // First drain the next writer
      //
      drainNextWriterDataSlot();

      //
      // Set new readerDataSlot - note, it may wrap
      //
      advanceWriterDataSlot();
   }

   void subscribe( TSubscription & s, int slot = InvalidSlot )
   {
      s.pRep  = NULL ;
      s.slot  = InvalidSlot ;

      //
      // Wait for the next frame to become available.
      //
      boost::unique_lock<boost::mutex> lock( newFrameMutex );
      while( readerDataSlot == slot )
      {
         try
         {
            boost::this_thread::interruption_point() ;
         }
         catch( ... )
         {
            lock.unlock();
            throw;
         }

         newFrameCond.wait(lock);
      }

      lock.unlock();

      //
      // Update number of readers for the current reader slot. Then record the dependency of
      // this client to the current reader slot.
      //
      TData & data = dataVector[ readerDataSlot ] ;

      //
      // atomic inc
      //
      data.numReaders++ ;

      //
      // Hand out a subscription.
      //
      s.pRep = data.pRep ;
      s.slot = readerDataSlot ;
   }

   void unsubscribe( TSubscription & s )
   {
      if( s.slot != InvalidSlot )
      {
         //
         // Release registration for this subscription
         //
         TData & data = dataVector[ s.slot ] ;

         XI_ASSERT_HARD( data.numReaders > 0 ) ;

         //
         // Decrease numReaders - then notify any waiting threads
         //
         {
            boost::lock_guard<boost::mutex> lock( unsubscribeMutex ) ;

            data.numReaders-- ;
         }

         unsubscribeCond.notify_all() ;
      }

      s.pRep = NULL ;
      s.slot = InvalidSlot ;
   }

   void renew( TSubscription & s )
   {
      int slot = s.slot ;

      unsubscribe( s ) ;
      subscribe( s, slot ) ;
   }

};

template<class REP>
class TTemporalDataQueue : public TDataQueue<REP>
{
public:

   TTemporalDataQueue( unsigned int N )
      : TDataQueue<REP>( N ) { } ;

   //
   // Note: This method is only to be called by the owning thread, aka the writer thread.
   //
   REP * operator[] (const int index)
   {
      //
      // [0] is the most current slot, [1] the next youngest, etc.
      //
      return this->dataVector[ ( this->writerDataSlot + index ) % this->N  ].pRep ;
   }
};

}}
#endif /* QUEUES_H_ */
